package com.atguigu.flink.sqlfunction;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Created by Smexy on 2023/3/5
 */
public class Demo9_SQLTopN
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        // 读取数据,产生水印
        String createTableSql = " create table t1 ( userId bigint, itemId bigint  , cId int ,  behavior string, ts bigint , " +
            "                   et  as TO_TIMESTAMP_LTZ(ts,0)    ,   " +
            "                   WATERMARK FOR et AS et - INTERVAL '1' SECOND   )" +
            " with ( " +
            " 'connector' = 'filesystem' ,   " +
            " 'path' =  'data/UserBehavior.csv' ,   " +
            "  'format' = 'csv' " +
            "      )                 ";

        tableEnvironment.executeSql(createTableSql);

        //过滤pv,按照itemid分组,第一次开窗聚合
        String sql1 = " select  window_start, window_end, itemId, count(*) click " +
            "  from TABLE( HOP(TABLE t1, DESCRIPTOR(et), INTERVAL '5' MINUTES , INTERVAL '1' HOURS) ) " +
            "  where  behavior = 'pv' " +
            "  GROUP BY window_start, window_end,itemId ";

        Table t2 = tableEnvironment.sqlQuery(sql1);
        tableEnvironment.createTemporaryView("t2",t2);

        /*
            第二次聚合,按照 窗口分组,求每一个 itemId的排名
                使用窗口函数 OverWindow
                    flink中只支持 row_number()函数

                order by 后只能写 时间字段。
                 OVER windows' ordering in stream mode must be defined on a time attribute.

               普通的Over窗口,order by后只能定义时间字段,如果希望定义非时间字段,必须是TopN的场景。

         */
        String sql2 = "select window_start, window_end, itemId, click ," +
                       " row_number() over( partition by  window_end order by click desc  )  rn" +
                      " from t2  ";

        Table t3 = tableEnvironment.sqlQuery(sql2);
        tableEnvironment.createTemporaryView("t3",t3);

        //使用 Flinksql的连接器,将结果写入到数据库
        /*
            创建一张表,映射到Mysql的 hot_item
            mysql是 TIMESTAMP ,flink中也是 TIMESTAMP
            mysql是bigint ,flink中也是bigint
            mysql是varchar,flink中是String

            流中有update类型,建的表就需要有主键。
         */
        String mysqlTable = "CREATE TABLE `t4` (" +
            "`w_start` TIMESTAMP ," +
            "  `w_end` TIMESTAMP ," +
            "  `item_id` BIGINT ," +
            "  `item_count` BIGINT," +
            "  `rk` BIGINT," +
            "  PRIMARY KEY (`w_end`,`rk`) NOT ENFORCED " +
            ") WITH (" +
            "   'connector' = 'jdbc'," +
            "   'url' = 'jdbc:mysql://hadoop104:3306/220926?useSSL=false'," +
            "   'table-name' = 'hot_item' ," +
            "   'username' = 'root' , " +
            "   'password' = '000000' " +
            ")";
        tableEnvironment.executeSql(mysqlTable);
        tableEnvironment.executeSql("insert into t4 select * from t3 where rn <= 3");



    }
}
